package com.ruleengine.plugin;

import com.ruleengine.destination.IDestinationProvider;
import com.ruleengine.plugin.util.ConfigUtils;
import com.ruleengine.util.model.Message;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.pf4j.Extension;

import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;

@Extension
public class KafkaDestinationProvider implements IDestinationProvider {
    private final KafkaProducer<String, String> kafkaProducer;

    public KafkaDestinationProvider() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                ConfigUtils.getKafkaProviderConfig().getBootstrapServers());
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        this.kafkaProducer = new KafkaProducer<>(properties);
    }

    @Override
    public CompletableFuture<Void> send(Message message) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        try {
            ProducerRecord<String, String> record
                    = new ProducerRecord<>(message.getMetadata().getUid(),
                    new String(message.getMessage(), StandardCharsets.UTF_8));
            kafkaProducer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    future.completeExceptionally(exception);
                }
                future.complete(null);
            });
        }catch (Exception exception) {
            future.completeExceptionally(exception);
        }
        return future;
    }

    @Override
    public void close() {
        kafkaProducer.close();
    }
}
